package wsz.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 公共的TP监控过滤器
 * @author wsz
 * @date 2021/12/15 13:58
 **/
@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter {

    private Map<String, List<Long>> timeCahceMap = new ConcurrentHashMap<>();

    class TimeTask implements Runnable {
        private final double TP_90 = 0.90;
        private final double TP_99 = 0.99;

        private long lastClear = -1;

        @Override
        public void run() {
            try {
                System.out.println("定时任务执行...");
                final Map<String, List<Long>> currentMap = timeCahceMap;
                for (Map.Entry<String, List<Long>> entry : currentMap.entrySet()) {
                    String key = entry.getKey();
                    List<Long> timeList = entry.getValue();
                    if (timeList.isEmpty()) continue;
                    // 排序
                    Collections.sort(timeList, Comparator.naturalOrder());
                    // 计算下标
                    int size = timeList.size();
                    int tp90 = (int) Math.floor(size * TP_90);
                    int tp99 = (int) Math.floor(size * TP_99);
                    System.out.println("key=" + key);
                    System.out.println("value(size="+timeList.size()+")=" + timeList);
                    System.out.println(key + ": TP90(index=" +tp90+")=" +timeList.get(tp90) + " TP99(index="+tp99+")=" + timeList.get(tp99));
                }
                if (lastClear < 0) {
                    lastClear = System.currentTimeMillis();
                    return;
                }
                // FIXME 只统计一分钟内数据，这样处理会大余一分钟才进行clear
                if ((System.currentTimeMillis() - lastClear) > 1000 * 60) {
                    timeCahceMap.clear();
                }
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    public TPMonitorFilter() {
        System.out.println("TPMonitorFilter");
        // 每隔5s进行打印
        ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutor.scheduleWithFixedDelay(new TimeTask(), 1, 5, TimeUnit.SECONDS);
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        long start = System.currentTimeMillis();
        String serviceName = invocation.getServiceName();
        String methodName = invocation.getMethodName();
        try {
            return invoker.invoke(invocation);
        } finally {
            String key = serviceName + "." + methodName;
            // key=serviceName.methodName
            List<Long> timeList = timeCahceMap.containsKey(key) ? timeCahceMap.get(key) : Collections.synchronizedList(new ArrayList<>());
            timeList.add(System.currentTimeMillis() - start);
            timeCahceMap.put(key, timeList);
        }
    }
}
